package com.juma.cartridge.retryque;

import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.DelayQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.atomic.AtomicBoolean;

import lombok.extern.slf4j.Slf4j;

import com.juma.cartridge.common.RetryExecutor;
import com.juma.cartridge.common.RetryUtils;
import com.juma.cartridge.retryque.cmd.Command;
import com.juma.cartridge.retryque.dao.RetryQueueDao;

@Slf4j
public class TaskScanner implements Runnable {

	private Conf conf;
	private DelayQueue<TaskLeaseProlong> leaseQueue;
	private RetryQueueDao retryQueueDao;
	private Map<String,RetryHandler> retryHandlers;
	private ExecutorService retryExecutor;
	private BlockingQueue<Command> commandQue;
	private AtomicBoolean running;
	
	public TaskScanner(Conf conf,DelayQueue<TaskLeaseProlong> leaseQueue,
			RetryQueueDao retryQueueDao,Map<String,RetryHandler> retryHandlers,
			ExecutorService retryExecutor,BlockingQueue<Command> commandQue) {
		this.conf = conf;
		this.leaseQueue = leaseQueue;
		this.retryQueueDao = retryQueueDao;
		this.retryHandlers = retryHandlers;
		this.retryExecutor = retryExecutor;
		this.commandQue = commandQue;
		this.running = new AtomicBoolean(false);
	}
	
	public boolean isRunning() {
	    return running.get();
	}
	
	@Override
	public void run() {
	    
	    if(!running.compareAndSet(false, true)) {
	        log.info("scanner正在执行,本次扫描跳过!");
	        return;
	    }
	    
		try {
			Long maxRound = retryQueueDao.getMaxRound(conf.getServiceId());
			//如果maxRound为null,说明表中没有可执行的任务项,直接返回
			if(maxRound==null) {
				return;
			}

			long currentRound = maxRound+1;
			log.debug("本轮[{}]任务扫描处理开始",currentRound);
			
			for(int i=1;!Thread.currentThread().isInterrupted() && processLoop(currentRound,i)>0;i++) {
			    log.debug("继续执行任务扫描处理[round={},idx={}]...",currentRound,i);
			}
			
			log.debug("本轮[{}]任务扫描处理结束!",currentRound);	
		}catch(Exception e) {
			log.error("扫描任务执行异常:"+e.getMessage(),e);
		}finally {
		    running.set(false);
		}
	}

	private int processLoop(long currentRound,int loopIdx) {
	    String scannerId = currentRound+"-"+loopIdx;
	    int zz = 1;
		int lockCount = 0;
		TaskLease lease = TaskLease.of(conf.getServiceId(),scannerId,conf.getLockSeconds(),currentRound);
		lease.setCount(conf.getLockCount());
		
		List<Long> releases = new LinkedList<Long>();
		try {
			//锁定任务
			lockCount = retryQueueDao.lock(lease);
			log.debug("[scanner={}]::本轮扫描锁定了{}条任务",scannerId,lockCount);
			if (lockCount==0) {
				return 0;
			}
			List<RetryTask> tasks = retryQueueDao.queryLocks(lease);
			if(tasks.isEmpty()) {
				return 0;
			}
			
			//加入续租队例
			TaskLeaseProlong prolong = TaskLeaseProlong.of(lease, tasks, conf.getLockSeconds()*1000);
			leaseQueue.add(prolong);
			
			List<RetryTask> updates = new LinkedList<RetryTask>();
			for (RetryTask task:tasks) {
				task.setRefresh(false);
				RetryHandler retryHandler = retryHandlers.get(task.getBizCode());
				if(retryHandler==null) {
					log.error("[scanner={}]::任务[id={},bizCode={}]没有注册对应的RetryHandler!!!",scannerId,task.getId(),task.getBizCode());
				}
				final TaskJob te = new TaskJob(retryHandler,task,scannerId,prolong,commandQue);
				//将task(包装为job)提交到线程池执行,若线程池资源溢出,则进行重试
				Boolean commited = RetryUtils.retry(new RetryExecutor<Boolean>(){
                    @Override
                    public Boolean execute() {
                        retryExecutor.execute(te);
                        return Boolean.TRUE;
                    }
				}, 20, 50);
				
				//如果task最终没有成功提交到线程池执行,则本次任务放弃执行,等到下轮扫描后重新提交
				if(commited!=Boolean.TRUE) {
				    releases.add(task.getId());
				    log.warn("工作线程超负载,本次任务提交执行失败,待下次轮询重试[taskId={}]", task.getId());
				}

				//如果当前扫出的任务中有round,retries值接近溢出时,需要将其值重置为0 (执行update即可置为0)
				//同时,本次扫描loop循环体需要结束,否则会导致任务重复扫描.
				int maxVal = Integer.MAX_VALUE-100;
				if(task.getRetries()>maxVal||task.getRound()>maxVal) {
				    //将zz值设为-1的作用,是使得本次loop结束后,不再进行下次loop.
				    zz = -1;
				    updates.add(task);
				}
			}
			if(!updates.isEmpty()) {
			    retryQueueDao.updateBatch(updates);
			}
			
			return lockCount * zz;
		}catch(Exception e) {
			log.error("[scanner={}]::本轮任务扫描处理异常中断",scannerId,e);
			return lockCount * zz;
		}finally {
		    if(!releases.isEmpty()) {
		        //释放本轮扫描未被成功执行的任务(释放后,下轮才扫得出来)
                final TaskFree free = TaskFree.of(scannerId, releases);
                RetryUtils.retry(new RetryExecutor<Boolean>(){
                    @Override
                    public Boolean execute() {
                        return Boolean.valueOf(retryQueueDao.release(free)>0);        
                    }
                }, 6, 500);
            }
		}
	}

}
